package com.rem.flink.flink6ProcessFunction;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * ProcessFunction 最基本的处理函数，基于 DataStream 直接调用.process()时作为参数传入。
 * KeyedProcessFunction 对流按键分区后的处理函数，基于 KeyedStream 调用.process()时作为参数传入。要想使用定时器，比如基于 KeyedStream。
 * ProcessWindowFunction 开窗之后的处理函数，也是全窗口函数的代表。基于 WindowedStream 调用.process()时作为参数传入。
 * ProcessAllWindowFunction 同样是开窗之后的处理函数，基于 AllWindowedStream 调用.process()时作为参数传入。
 * CoProcessFunction 合并（connect）两条流之后的处理函数，基于 ConnectedStreams 调用.process()时作为参数传入。
 * ProcessJoinFunction 间隔连接（interval join）两条流之后的处理函数，基于 IntervalJoined 调用.process()时作为参数传入
 * BroadcastProcessFunction 广播连接流处理函数，基于 BroadcastConnectedStream 调用.process()时作为参数传入。是一个未keyBy 的普通 DataStream 与一个广播流（BroadcastStream）做连接（conncet）之后的产物
 * KeyedBroadcastProcessFunction 按键分区的广播连接流处理函数，同样是基于 BroadcastConnectedStream 调用.process()时作为参数传入。与 BroadcastProcessFunction 不同的是，这时的广播连接流，是一个 KeyedStream与广播流（BroadcastStream）做连接之后的产物。
 * <p>
 * <p>
 * ProcessFunction demo
 *
 * @author Rem
 * @date 2022-10-13
 */

public class ProcessFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));


        /**
         * 最基础 基于 DataStream 后面调用
         * mary出现打印一次 Bob出现打印两次
         */
        stream.process(new ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
                if ("Mary".equals(value.getUser())) {
                    out.collect(value.getUser());
                } else if ("Bob".equals(value.getUser())) {
                    out.collect(value.getUser());
                    out.collect(value.getUser());
                }
                System.out.println(ctx.timerService().currentWatermark());


                /**
                 * 定时服务
                 */
//                TimerService timerService = ctx.timerService();
//                // 获取当前的处理时间
//                timerService.currentProcessingTime();
//                // 注册处理时间定时器，当处理时间超过 time 时触发
//                timerService.registerProcessingTimeTimer(5);
//                // 删除触发时间为 time 的处理时间定时器
//                timerService.deleteProcessingTimeTimer(5);
//
//
//                // 获取当前的水位线（事件时间）
//                timerService.currentWatermark();
//                // 注册事件时间定时器，当水位线超过 time 时触发
//                timerService.registerEventTimeTimer(5);
//                // 删除事件时间为 time 的处理时间定时器
//                timerService.deleteEventTimeTimer(5);

            }

        }).print();

        env.execute();

    }


}
